package com.qdc.datacenters.factory;

import cn.hutool.core.thread.ThreadUtil;
import com.alibaba.fastjson.JSONObject;
import com.qdc.datacenters.constant.ChatMsgConstants;
import com.qdc.datacenters.domain.queue.ChatMsgQueue;
import com.qdc.datacenters.handler.ConsumerHandler;
import com.qdc.datacenters.service.impl.ThreadFactoryImpl;
import com.qdc.datacenters.utils.ThreadUtils;
import jakarta.annotation.PreDestroy;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @author danmo
 * @date 2024-02-29 16:55
 **/
@Component
@Slf4j
public class QwChatMsgRouteFactory implements ApplicationRunner {
    private final Map<String, ConsumerHandler> routeMap = new ConcurrentHashMap<>();

    private final ThreadPoolExecutor executorService = new ThreadPoolExecutor(1, 1, 10L, TimeUnit.MINUTES,
            new LinkedBlockingQueue<Runnable>(), new ThreadFactoryImpl("QwChatMsgRouteFactoryThread_",true));


    public static volatile boolean flag = false;

    public QwChatMsgRouteFactory(Map<String, ConsumerHandler> routeMap) {
        routeMap.putAll(routeMap);
    }

    public void register(String key, ConsumerHandler routeHandler) {
        routeMap.put(key, routeHandler);
    }


    public void start() {
        executorService.execute(() -> {
            for (;;) {
                if(!flag || ChatMsgConstants.chatMsgQueues.isEmpty()){
                    ThreadUtil.safeSleep(5*1000);
                    continue;
                }
                ChatMsgQueue queue = ChatMsgConstants.chatMsgQueues.poll();
                for (ConsumerHandler handler : routeMap.values()) {
                    try {
                        handler.handler(queue);
                    } catch (Exception e) {
                        log.error("异常消费Queue:{},ex:{}", JSONObject.toJSON(queue), e.getMessage(), e);
                    }
                }
            }
        });
    }

    @PreDestroy
    public void shutdown() {
        ThreadUtils.shutdownAndAwaitTermination(executorService);
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        start();
    }
}
